package com.xiashitech.apiservice.grpc;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.xiashitech.apiservice.dto.AgentIdentityDTO;
import com.xiashitech.interfaces.grpc.*;
import io.grpc.stub.ServerCallStreamObserver;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
//@Service
@GrpcService
public class IncrementalCacheService extends RPCIncrementalCacheServiceGrpc.RPCIncrementalCacheServiceImplBase {
    private final ConcurrentHashMap<String, List<StreamObserver<RPCIncrementalCacheResponse>>> responseAuthFactMap = new ConcurrentHashMap<>();
    // TODO 怎么没有心跳？可能是 ConfigService 中有心跳了，是同一个tcp链接，这个可能不用心跳了，待验证
    // TODO FOR DEBUG
//    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
//
//    private IncrementalCacheService () {
//        scheduledExecutorService.scheduleWithFixedDelay(()->{
//            try {
//                Random r = new Random();
//                int n = r.nextInt(200);
//                log.info("random int: {}", n);
//                if(n != 50)
//                    return;
//
//                responseAuthFactMap.entrySet().stream().forEach(entry->{
//                    String mapkey = entry.getKey();
//                    log.info("now mapkey: " + mapkey);
//                    String[] keyPart = mapkey.split("/");
//                    int len = keyPart.length;
//                    if(len >= 3) {
//                        log.info("key split >= 3: {}-{}-{}", keyPart[0],keyPart[1],keyPart[2]);
//                        pushIncrementToClient(keyPart[0],keyPart[1],keyPart[2], "testtest-" + String.valueOf(new Date()));
//                    } else if(len >= 2) {
//                        log.info("key split >= 2: {}-{}", keyPart[0],keyPart[1]);
//                        pushIncrementToClient(keyPart[0],keyPart[1], "", "testtest-" + String.valueOf(new Date()));
//                    } else {
//                        log.info("key split else: {}, {}", keyPart[0], mapkey);
//                        pushIncrementToClient(keyPart[0],"","", "testtest-" + String.valueOf(new Date()));
//                    }
//                });
//            } catch (Exception e) {
//                e.printStackTrace();
//            }
//        }, 3, 3, TimeUnit.SECONDS) ;
//    }
    @Override
    public StreamObserver<RPCIncrementalCacheRequest> biStreamIncrementalCache(StreamObserver<RPCIncrementalCacheResponse> responseObserver) {
        StreamObserver<RPCIncrementalCacheRequest> req = new StreamObserver<RPCIncrementalCacheRequest>() {
            @Override
            public void onNext(RPCIncrementalCacheRequest value) {
                String reqJson = value.getReqJson();
                if(reqJson == null || "".equals(reqJson)) {
                    log.info("请求错误，必须有内容，如指定 SystemID、心跳消息");
                    return;
                }
                if("INCREMENTHEARTBEAT".equals(reqJson)) {
                    log.info("biStreamIncrementalCache 收到心跳 HEARTBEAT，只给出心跳应答");
                    RPCIncrementalCacheResponse resp = RPCIncrementalCacheResponse.newBuilder().setRespJson("INCREMENTHEARTBEAT").build();
                    synchronized (responseObserver) {
                        responseObserver.onNext(resp);
                    }
                    return;
                }
                // {"systemId": "", "serviceId": "", "serialId": ""}，收到内容，构建 map ，不响应内容
                AgentIdentityDTO agentId = JSONUtil.toBean(reqJson, AgentIdentityDTO.class);
                String systemId = agentId.getSystemId();
                String serviceId = agentId.getServiceId();
                String serialId = agentId.getSerialId();
                log.info("biStreamIncrementalCache, receive the client regist content: {}, {}, {}, no response.", systemId, serviceId,serialId);
                // 缓存 responseObserver ，用于推送
                synchronized (responseAuthFactMap) {
                    String key = buildKey(systemId, serviceId, serialId); // 按理这能构成了唯一key，为避免误配置，保留原数据结构，支持三者组成的key不唯一
                    List<StreamObserver<RPCIncrementalCacheResponse>> respList = responseAuthFactMap.get(key);
                    if(respList == null) {
                        respList = new ArrayList<>();
                        respList.add(responseObserver);
                        responseAuthFactMap.put(key, respList);
                    } else {
                        if(!respList.stream().anyMatch(item->item == responseObserver))
                            respList.add(responseObserver);
                    }
                }
            }

            @Override
            public void onError(Throwable t) {
                // 不做 map 遍历了，推送时检查，无效再删除
                log.info("biStreamIncrementalCache, onError: {}", t.getMessage());
            }

            @Override
            public void onCompleted() {
                log.info("biStreamIncrementalCache, onComplete");
            }
        };

        return req;
    }

    private String buildKey(String systemId, String serviceId, String serialId) {
        if(null == systemId || "".equals(systemId)) {
            log.error("buildKey err: systemId can't empty");
            return "";
        }
        if(null == serviceId || "".equals(serviceId) || null == serialId || "".equals(serialId)) {
            return systemId.toLowerCase() + "/";
        }
        return systemId.toLowerCase() + "/" + serviceId.toLowerCase() + "/" + serialId.toLowerCase();
    }
    public List<AgentIdentityDTO> getAgentIds(String systemId) {
        List<AgentIdentityDTO> keys = new ArrayList<>();
        synchronized (responseAuthFactMap) {
           keys = responseAuthFactMap.entrySet().stream()
               .filter(entry -> entry.getValue().size() > 0 && entry.getKey().startsWith(systemId+"/"))
               .map(entry-> {
                   // System.out.println("getAgentIds entry key: " + entry.getKey());
                   String [] keyPart = entry.getKey().split("/");
                   String localSystemId = keyPart[0];
                   String serviceId = "";
                   String serialId = "";
                   if(keyPart.length >= 1) {
                       serviceId = keyPart[1];
                   }
                   if( keyPart.length > 1) {
                       serialId = keyPart[2];
                   }
                   AgentIdentityDTO ai = AgentIdentityDTO.builder()
                           .systemId(localSystemId)
                           .serviceId(serviceId)
                           .serialId(serialId)
                           .build();
                   return ai;
           }).collect(Collectors.toList());
        }
        return keys;
    }

    public List<AgentIdentityDTO> getAllAgentIds() {
        List<AgentIdentityDTO> keys = new ArrayList<>();
        synchronized (responseAuthFactMap) {
            keys = responseAuthFactMap.entrySet().stream()
                    .filter(entry -> entry.getValue().size() > 0)
                    .map(entry-> {
                        // System.out.println("getAllAgentIds entry key: " + entry.getKey());
                        String [] keyPart = entry.getKey().split("/");
                        String systemId = keyPart[0];
                        String serviceId = "";
                        String serialId = "";
                        if(keyPart.length >= 1) {
                            serviceId = keyPart[1];
                        }
                        if( keyPart.length > 1) {
                            serialId = keyPart[2];
                        }
                        AgentIdentityDTO ai = AgentIdentityDTO.builder()
                                .systemId(systemId)
                                .serviceId(serviceId)
                                .serialId(serialId)
                                .build();
                        return ai;
                    }).collect(Collectors.toList());
        }
        return keys;
    }

    /**
     * 推送缓存数据到客户端
     * @param systemId
     * @param content
     * @return false：没有推送，true：推送了，但不确认推送成功
     */
    public void pushIncrementToClient(String systemId, String serviceId, String serialId, String content) {
        if(null==serviceId || "".equals(serviceId) || null==serialId || "".equals(serialId)) {
            pushIncrementToClient(systemId, content);
            return;
        }
        String key = buildKey(systemId, serviceId, serialId);
        // log.info("pushIncrementToClient enter, key: " + key + ", content length: " + content.length());
        List<StreamObserver<RPCIncrementalCacheResponse>> respList = responseAuthFactMap.get(key);
        if(respList == null || respList.size() == 0){
            // log.info("pushIncrementToClient responseAuthFactMap is empty, key: " + key);
            return;
        }
        // log.info("pushIncrementToClient responseAuthFactMap key: {}, list size: ", key, respList.size());
        synchronized (responseAuthFactMap) { // 为避免并发对对 list 的删除，加锁
            Iterator<StreamObserver<RPCIncrementalCacheResponse>> iterator = respList.iterator();
            int idx = 0;
            while (iterator.hasNext()) {
                StreamObserver<RPCIncrementalCacheResponse> responseObserver = iterator.next();
                ServerCallStreamObserver serverCallStreamObserver = (ServerCallStreamObserver) responseObserver;
                if (serverCallStreamObserver.isCancelled()) {
                    log.info("pushIncrementToClient remove: " + key + ": " + serverCallStreamObserver);
                    iterator.remove();
                } else {
                    idx++;
                    log.info("pushIncrementToClient key: {}, push idx: {}", key, idx);
                    RPCIncrementalCacheResponse resp = RPCIncrementalCacheResponse.newBuilder().setRespJson(content).build();
                    synchronized (responseObserver) {
                        responseObserver.onNext(resp);
                    }
                }
            }
            if(respList.size() == 0) { // 如果 map 下的列表空了，删除 map entry
                responseAuthFactMap.remove(key);
            }
        }
    }
    public void pushIncrementToClient(String systemId, String content) {
        String key = buildKey(systemId, null, null);
        // log.info("pushIncrementToClient enter, key: " + key + ", content length: " + content.length());
        synchronized (responseAuthFactMap) { // 为避免并发对对 list 的删除，加锁
            Iterator<Map.Entry<String, List<StreamObserver<RPCIncrementalCacheResponse>> >> entryIterator = responseAuthFactMap.entrySet().iterator();
            while(entryIterator.hasNext()) {
                Map.Entry<String, List<StreamObserver<RPCIncrementalCacheResponse>> > entry = entryIterator.next();
                if(!entry.getKey().startsWith(key))
                    continue;
                List<StreamObserver<RPCIncrementalCacheResponse>> respList = entry.getValue();
                Iterator<StreamObserver<RPCIncrementalCacheResponse>> iterator = respList.iterator();
                int idx = 0;
                while (iterator.hasNext()) {
                    StreamObserver<RPCIncrementalCacheResponse> responseObserver = iterator.next();
                    ServerCallStreamObserver serverCallStreamObserver = (ServerCallStreamObserver) responseObserver;
                    if (serverCallStreamObserver.isCancelled()) {
                        log.info("pushIncrementToClient remove: " + key + ": " + serverCallStreamObserver);
                        iterator.remove();
                    } else {
                        idx++;
                        log.info("pushIncrementToClient key: {}, push idx: {}", key, idx);
                        RPCIncrementalCacheResponse resp = RPCIncrementalCacheResponse.newBuilder().setRespJson(content).build();
                        synchronized (responseObserver) {
                            responseObserver.onNext(resp);
                        }
                    }
                }
                if (respList.size() == 0) { // 如果 map 下的列表空了，删除 map entry
                    entryIterator.remove();
                }
            }
        }
    }
}
